﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;

namespace ConfirmPublisher
{
    internal class Program
    {
        static ConcurrentDictionary<ulong, string> outstandingConfirms = new ConcurrentDictionary<ulong, string>();

        static void Main(string[] args)
        {
            try
            {
                var factory = new ConnectionFactory() { HostName = "localhost" };
                using var connection = factory.CreateConnection();
                using var channel = connection.CreateModel();
                channel.ConfirmSelect();
                channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null);
                channel.BasicAcks += (sender, ea) =>
                {
                    Console.WriteLine("消息已经被确认");
                    Console.WriteLine($"Tag: {ea.DeliveryTag}");
                    Console.WriteLine($"Multiple: {ea.Multiple}");
                    CleanOutstandingConfirms(ea.DeliveryTag, ea.Multiple);
                };

                channel.BasicNacks += (sender, ea) =>
                {
                    Console.WriteLine("消息没有被确认");
                    outstandingConfirms.TryGetValue(ea.DeliveryTag, out string body);
                    //Console.WriteLine($"Tag: {ea.DeliveryTag}");
                    //Console.WriteLine($"Multiple: {ea.Multiple}");
                    //Console.WriteLine($"Requeue: {ea.Requeue}");
                    Console.WriteLine($"Message with body {body} has been nack-ed. Sequence number: {ea.DeliveryTag}, multiple: {ea.Multiple}");
                    CleanOutstandingConfirms(ea.DeliveryTag, ea.Multiple);
                };

                

                //var batchSize = 100;
                //var outstandingMessageCount = 0;
                for (int i = 1; i <= 50000; i++)
                {
                    var msg = $"message{i}";
                    var body = Encoding.UTF8.GetBytes(msg);
                    outstandingConfirms.TryAdd(channel.NextPublishSeqNo, msg);
                    channel.BasicPublish(exchange: string.Empty, routingKey: "hello", basicProperties: null, body: body);
                    //outstandingMessageCount++;
                    //if (outstandingMessageCount == batchSize)
                    //{
                    //    channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));
                    //    Console.WriteLine("本轮等待确认结束");
                    //    outstandingMessageCount = 0;
                    //}
                }

                Console.ReadLine();
            }
            catch (Exception ex)
            {
                throw;
            }
        }

        static void CleanOutstandingConfirms(ulong sequenceNumber, bool multiple)
        {
            if (multiple)
            {
                var confirmed = outstandingConfirms.Where(k => k.Key <= sequenceNumber);
                foreach (var entry in confirmed)
                {
                    outstandingConfirms.TryRemove(entry.Key, out _);
                }
            }
            else
            {
                outstandingConfirms.TryRemove(sequenceNumber, out _);
            }
        }
    }
}